package com.lz.hbase;

/**
 * @description
 *
 * @date 2021-11-25 15:45
 *
 * @author Lizhong
 */

import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author arvin
 */
public class HbaseTest {
    private static Admin admin;

    static final String COLUMNS_FAMILY_1 = "cf1";
    static final String COLUMNS_FAMILY_2 = "cf2";

    public static Connection initHbase() throws IOException {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "c8c56bb6c0d1");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        configuration.set("hbase.master", "c8c56bb6c0d1:60000");
        Connection connection = ConnectionFactory.createConnection(configuration);
        return connection;
    }

    //创建表 create
    public static void createTable(TableName tableName, String[] cols) throws IOException {
        admin = initHbase().getAdmin();
        if (admin.tableExists(tableName)) {
            admin.enableTable(tableName);
            System.out.println("Table Already Exists！");
        } else {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
            for (String col : cols) {
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(col);
                hTableDescriptor.addFamily(hColumnDescriptor);
            }
            admin.createTable(hTableDescriptor);
            System.out.println("Table Create Successful");
        }
    }

    public static TableName getTbName(String tableName) {
        return TableName.valueOf(tableName);
    }

    // 删除表 drop
    public static void deleteTable(TableName tableName) throws IOException {
        admin = initHbase().getAdmin();
        if (admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            System.out.println("Table Delete Successful");
        } else {
            System.out.println("Table does not exist!");
        }

    }

    //put 插入数据
    public static void insertData(TableName tableName, IOT iot) throws IOException {
        Put put = new Put(Bytes.toBytes(iot.getId()));
        put.addColumn(Bytes.toBytes(COLUMNS_FAMILY_1), Bytes.toBytes("value"), Bytes.toBytes(iot.getValue()));
        put.addColumn(Bytes.toBytes(COLUMNS_FAMILY_1), Bytes.toBytes("type"), Bytes.toBytes(iot.getType0()));
        put.addColumn(Bytes.toBytes(COLUMNS_FAMILY_1), Bytes.toBytes("time"), Bytes.toBytes(iot.getTime()));
        initHbase().getTable(tableName).put(put);
        System.out.println("Data insert success:");
    }

    public static void batchInsertData(TableName tableName, List<IOT> iots) throws IOException {
        List<Put> puts = iots.stream().map(iot -> {
            Put put = new Put(Bytes.toBytes(iot.getId()));
            put.addColumn(Bytes.toBytes(COLUMNS_FAMILY_1), Bytes.toBytes("value"), Bytes.toBytes(iot.getValue()));
            put.addColumn(Bytes.toBytes(COLUMNS_FAMILY_1), Bytes.toBytes("type0"), Bytes.toBytes(iot.getType0()));
            put.addColumn(Bytes.toBytes(COLUMNS_FAMILY_1), Bytes.toBytes("time"), Bytes.toBytes(iot.getTime()));
            put.addColumn(Bytes.toBytes(COLUMNS_FAMILY_1), Bytes.toBytes("deviceId"), Bytes.toBytes(iot.getDeviceId()));
            return put;
        }).collect(Collectors.toList());
        initHbase().getTable(tableName).put(puts);
        System.out.println("Data insert success:");
    }

    // delete 删除数据
    public static void deleteData(TableName tableName, String rowKey) throws IOException {
        Delete delete = new Delete(Bytes.toBytes(rowKey));      // 指定rowKey
//        delete = delete.addColumn(Bytes.toBytes(COLUMNS_FAMILY_1), Bytes.toBytes("name"));  // 指定column，也可以不指定，删除该rowKey的所有column
        initHbase().getTable(tableName).delete(delete);
        System.out.println("Delete Success");
    }

    // scan数据
    public static List<JSONObject> scan(TableName tableName , String col, String value) throws IOException {
        Scan scan = new Scan();
        byte[] cf = Bytes.toBytes("cf1");
        scan.addFamily(cf);
        scan.setFilter(new FilterList( new RowFilter( CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("1637896702963"))))

        );

        ResultScanner results = initHbase().getTable(tableName).getScanner(scan);

        List<JSONObject> list = new ArrayList<>();
        for (Result result : results) {
            Cell[] cells = result.rawCells();
            JSONObject map = new JSONObject();
            map.put("id",  Bytes.toString(result.getRow()));
            for (Cell cell : cells) {
                String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                String val = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                map.put(colName,val);
            }
            list.add(map);
        }
        return list;
    }

    // scan数据
    public static List<Student> allScan(TableName tableName) throws IOException {
        ResultScanner results = initHbase().getTable(tableName).getScanner(new Scan().addFamily(Bytes.toBytes("cf1")));
        List<String> list = new ArrayList<>();
        for (Result result : results) {
            Student student = new Student();

        }
        return null;
    }

    // 根据rowkey get数据
    public static JSONObject singleGet(TableName tableName, String rowKey) throws IOException {
        JSONObject student = new JSONObject();
        Get get = new Get(Bytes.toBytes(rowKey));
        if (!get.isCheckExistenceOnly()) {
            Result result = initHbase().getTable(tableName).get(get);
            for (Cell cell : result.rawCells()) {
                String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                student.put(colName, value);

            }

        }

        System.out.println(student.toString());
        return student;
    }

    // 查询指定Cell数据
    public static String getCell(TableName tableName, String rowKey, String cf, String column) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        String rst = null;
        if (!get.isCheckExistenceOnly()) {
            get = get.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column));
            try {

                Result result = initHbase().getTable(tableName).get(get);
                byte[] resByte = result.getValue(Bytes.toBytes(cf), Bytes.toBytes(column));
                rst = Bytes.toString(resByte);
            } catch (Exception exception) {
                System.out.printf("columnFamily or column does not exists");
            }

        }
        System.out.println("Value is: " + rst);
        return rst;
    }

    public static List<JSONObject> getALl(TableName tableName, String cf) throws IOException {


        ResultScanner results = initHbase().getTable(tableName).getScanner(Bytes.toBytes(cf));

        Iterator<Result> iterator = results.iterator();
        List<JSONObject> ret= new ArrayList<>();
        while (iterator.hasNext()) {
            Result result = iterator.next();
            // byte[] resByte = result.getRow();
            // String s = Bytes.toString(resByte);
            JSONObject json = new JSONObject();
            byte[] row = result.getRow();
            json.put("id",Bytes.toString(row));
            for (Cell cell : result.rawCells()) {
                String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                json.put(colName,value);
            }
            ret.add(json);
        }
        return ret;
    }


    public static void main(String[] args) throws IOException {
        Student student = new Student();
        student.setId("1");
        student.setName("Arvin");
        student.setAge("18");
        student.setHobby("basketball");
        String table = "student";

        createTable(getTbName(table), new String[]{COLUMNS_FAMILY_1, COLUMNS_FAMILY_2});
//        deleteTable(getTbName(table));
//        insertData(getTbName(table), student);
        // deleteData(getTbName(table), "1");
        singleGet(getTbName(table), "2");
        singleGet(getTbName(table), "1");
        getCell(getTbName(table), "1", "cf1", "name");
    }
}
